Description

Implements a subset of the Prometheus metrics exposed by Celery Flower directly in the Celery instrumentation, so users who only need basic task/worker observability no longer need to run Flower as a separate service.

Fixes #3458

Type of change

• New feature (non-breaking change which adds functionality)
• Bug fix (non-breaking change which fixes an issue)

Changes

Task-level metrics (in CeleryInstrumentor)

Metric tracking wired into the existing prerun/postrun/failure/retry handlers:

Worker lifecycle metrics (new CeleryWorkerInstrumentor)

A separate instrumentor that hooks into worker_ready / worker_shutdown / task_received / task_revoked signals in the main worker process:

Registered as a new celery_worker entry point in pyproject.toml.

Omitted metrics

flower_task_prefetch_time_seconds and flower_worker_prefetched_tasks are not implemented. In pool=prefork mode (the default), task_received fires in the main process while task_prerun fires in a child process — there is no shared state to compute the time delta between them. Flower can do this because it consumes Celery events externally in a single process; replicating that inside the worker is not feasible without producer-side instrumentation or protocol changes.

Bug fix — memory leak in task_id_to_start_time

Previously, task_id_to_start_time was a class-level dict that was never cleaned up after task completion, growing unbounded over time. It is now instance-scoped and entries are removed in _trace_postrun.

Housekeeping

• Type annotations added across __init__.py and utils.py
• Typed CeleryGetter properly (Getter[Request])
• Null guards added to detach_context / retrieve_context / retrieve_task_id_from_message in utils.py
• Replaced bare dict metrics store with typed _CeleryTaskMetrics / _CeleryWorkerMetrics dataclasses

How Has This Been Tested?

• Comprehensive unit tests covering all new signal handlers and metric emissions (test_metrics.py, ~1100 new lines)
• Integration tests that start a real Celery worker in-process to verify worker_ready / worker_shutdown signals
• Edge-case tests: duplicate signals, unknown workers, uninstrument/reinstrument cycles
• Pre-commit hooks (ruff lint + format) passing
• End-to-end testing with a real Django + Celery project exporting to Grafana via OTLP

Trying it out

A self-contained example project (Django + Celery + Docker Compose + Grafana LGTM) is available at celery-metrics-example (separate branch on the fork, not part of this PR).

Quick start:

git clone -b celery-metrics-example https://github.com/diogosilva30/opentelemetry-python-contrib.git /tmp/celery-example
cd /tmp/celery-example/instrumentation/opentelemetry-instrumentation-celery/example
docker compose up -d

# Exercise every metric path
docker compose run --rm -w /repo django \
    uv run --no-sync \
        --directory instrumentation/opentelemetry-instrumentation-celery/example \
        python exercise_metrics.py

Then open Grafana at http://localhost:3000 (admin/admin) and explore metrics prefixed with flower.*.

Does This PR Require a Core Repo Change?

• No.

Checklist:

• Followed the style guidelines of this project
• Changelogs have been updated
• Unit tests have been added
• Documentation has been updated